package com.songmy.spider.run.my;


import java.io.Closeable;
import java.io.IOException;
import java.util.LinkedList;
import java.util.Queue;

import org.apache.log4j.Logger;

import com.alibaba.fastjson.JSONObject;
import com.songmy.spider.run.callback.CallBackFace;
import com.songmy.spider.run.callback.impl.FileBack;
import com.songmy.spider.run.callback.impl.NormalBack;
import com.songmy.spider.run.vo.SpiderRule;

import us.codecraft.webmagic.ResultItems;
import us.codecraft.webmagic.Task;
import us.codecraft.webmagic.pipeline.Pipeline;

/***
 * 回调内容
 * @author Marc
 * @version 1.0
 */
public class CallBackPipeline implements Pipeline,Closeable,Runnable {
	SpiderRule rule = null ;
	CallBackFace call = null ; 
	Queue<JSONObject> queue = new LinkedList<JSONObject>();
	Logger log = Logger.getLogger(getClass());
	Thread t = null ; 
	
	public CallBackPipeline(SpiderRule rule ) {
		this.rule = rule ; 
		switch (this.rule.getCallBack().getCallBackType()) {
		case FILE:
			call = new FileBack();
			break;

		default:
			call = new NormalBack();
			break;
		}
		t = new Thread(this);
		t.setName("Thread-CallBack");
	}
	
	final static String _VALUE = "_VALUE";
	
	public void process(ResultItems resultItems, Task task) {
		queue.add((JSONObject)resultItems.get(_VALUE));
		if(!t.isInterrupted() && !t.isAlive()){
			synchronized (task) {
				if(!t.isInterrupted() && !t.isAlive()){
					t.start();
				}
			}
		}
	}

	public void close() throws IOException {
		if(call!=null)
			call.stop();
		t.stop();
		
	}

	public void run() {
		JSONObject o = null ; 
		while (!Thread.currentThread().isInterrupted() ) {
			o = queue.poll();
			if(o!=null){
				call.setValue(o,this.rule.getCallBack());
				call.call();
			}else{
				waitJson();
			}
		}
	}

	private void waitJson() {
		try {
			log.debug(" wait call something");
			if( (!Thread.currentThread().isInterrupted() && t.isAlive() )){
				Thread.sleep(300);
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	
	
	

}
